package com.zliuyno.rpcdemo;







import com.zliuyno.io.testreactor.netty.SerDerUtil;
import com.zliuyno.rpcdemo.proxy.MyProxy;
import com.zliuyno.rpcdemo.rpc.Dispatcher;
import com.zliuyno.rpcdemo.rpc.Protocol.MyContent;
import com.zliuyno.rpcdemo.rpc.Protocol.MyHeader;
import com.zliuyno.rpcdemo.rpc.ResponseMappingCallback;
import com.zliuyno.rpcdemo.service.*;
import com.zliuyno.rpcdemo.transport.ClientFactory;
import com.zliuyno.rpcdemo.transport.MyHttpRpcHandler;
import com.zliuyno.rpcdemo.transport.ServerDecode;
import com.zliuyno.rpcdemo.transport.ServerRequestHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.junit.Test;

import javax.servlet.ServletException;
import javax.servlet.ServletInputStream;
import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.*;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;

public class MyRPCTest {
    @Test
    public void get() {

        AtomicInteger num = new AtomicInteger(0);

       /* new Thread(() -> {
            startServer();
        }).start();

        System.out.println("server started...");*/

        int size = 50;
        Thread[] threads = new Thread[size];
        for (int i = 0; i < size; i++) {
            threads[i] = new Thread(() -> {
                Car car = MyProxy.proxyGet(Car.class);   //动态代里实现
                String arg = "Hello" + num.incrementAndGet();
                String res = car.ooxx(arg);
                System.out.println("client over msg >> " + res + " -- src arg - " + arg);
            });
        }

        for (Thread thread : threads) {
            thread.start();
        }

        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }

//        Fly fly = proxyGet(Fly.class);   //动态代里实现
//        fly.xxoo("Hello");
    }

    @Test
    public void startServer() {

        Car myCar = new MyCar();
        Fly myFly = new MyFly();
        Dispatcher dis = Dispatcher.getDis();
        dis.register(Car.class.getName(), myCar);
        dis.register(Fly.class.getName(), myFly);

        NioEventLoopGroup boss = new NioEventLoopGroup(20);
        NioEventLoopGroup worker = boss;

        ServerBootstrap bs = new ServerBootstrap();
        ChannelFuture bind = bs.group(boss, worker)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        System.out.println("server accept client port :" + ch.remoteAddress().getPort());
                        ChannelPipeline pipeline = ch.pipeline();

                        //1.自定义的rpc,自定义的协议需要关注那些问题：粘包拆包问题 header+body
                        /*pipeline.addLast(new ServerDecode());
                        pipeline.addLast(new ServerRequestHandler(dis));*/
                        //2.小火车 传输协议用的就是http了--你可以自己学字节码byte[]
                        //其实netty提供了一套编解码
                        pipeline.addLast(new HttpServerCodec())
                                .addLast(new HttpObjectAggregator(1024 * 512))
                                .addLast(new ChannelInboundHandlerAdapter() {
                                    @Override
                                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                        //http协议,这个msg是啥：完整的http-request
                                        FullHttpRequest request = (FullHttpRequest) msg;
                                        System.out.println(request.toString()); //因为现在consumer使用的是一个现成的URL

                                        //这个就是consumer 序列化的MyContent
                                        ByteBuf content = request.content();
                                        byte[] data = new byte[content.readableBytes()];
                                        content.readBytes(data);
                                        ObjectInputStream oin = new ObjectInputStream(new ByteArrayInputStream(data));
                                        MyContent myContent = (MyContent) oin.readObject();

                                        String serviceName = myContent.getName();
                                        String methodName = myContent.getMethodName();
                                        Object c = dis.get(serviceName);
                                        Class<?> clazz = c.getClass();
                                        Method method = clazz.getMethod(methodName, myContent.getParameterTypes());
                                        Object res = method.invoke(c, myContent.getArgs());
                                        String execThreadName = Thread.currentThread().getName();

                                        MyContent resContent = new MyContent();
                                        resContent.setRes(res);
                                        byte[] contentByte = SerDerUtil.ser(resContent);

                                        DefaultFullHttpResponse response = new DefaultFullHttpResponse
                                                (HttpVersion.HTTP_1_0, HttpResponseStatus.OK, Unpooled.copiedBuffer(contentByte));

                                        response.headers().set(HttpHeaderNames.CONTENT_LENGTH,contentByte.length);

                                        //http协议,header+body
                                        ctx.writeAndFlush(response);

                                    }
                                });
                    }
                }).bind(new InetSocketAddress("localhost", 9090));
        try {
            bind.sync().channel().closeFuture().sync();

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void startHttpServer(){

        Car myCar = new MyCar();
        Fly myFly = new MyFly();
        Dispatcher dis = Dispatcher.getDis();
        dis.register(Car.class.getName(), myCar);
        dis.register(Fly.class.getName(), myFly);

        //tomcat jetty [servlet]
        Server server = new Server(new InetSocketAddress("localhost",9090));
        ServletContextHandler handler = new ServletContextHandler(server, "/");
        server.setHandler(handler);
        handler.addServlet(MyHttpRpcHandler.class,"/*");
        try {
            server.start();
            server.join();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Test
    public void testRPC() {

        Car car = MyProxy.proxyGet(Car.class);
        Persion zhangsan = car.oxox("zhangsan", 16);
        System.out.println(zhangsan);
    }


}











